package consumer

import (
	"album/dao"
	"album/internal/svc"
	"album/repo"
	"context"
	"github.com/IBM/sarama"
	"go.uber.org/zap"
	"listenboos-common/kafka/model"
	"search/kafka/saramax"
)

type TrackConsumer struct {
	svcCtx   *svc.ServiceContext
	client   sarama.Client
	trackDao repo.TrackInfoRepo
	//albumSearchDao *dao.AlbumElasticDAO
}

func NewTrackConsumer(
	svcCtx *svc.ServiceContext,
	client sarama.Client,
) *TrackConsumer {
	return &TrackConsumer{
		client:   client,
		trackDao: dao.NewTrackInfoDao(svcCtx.DB),
		//	albumSearchDao: dao.NewAlbumElasticDAO(),
	}
}
func (r *TrackConsumer) Start() error {
	cg, err := sarama.NewConsumerGroupFromClient("track_collect",
		r.client)
	if err != nil {
		zap.L().Error("im系统kafka消息记录消费错误", zap.Error(err))
		return err
	}
	go func() {
		err = cg.Consume(context.Background(),
			[]string{"track_collect"},
			saramax.NewHandler[model.TrackModel](r.TrackConsumer))
		if err != nil {
			// 记录日志
			zap.L().Error("kafka消息记录消费错误", zap.Error(err))
			//	return
		}
	}()
	return err
}
func (r *TrackConsumer) TrackConsumer(msg *sarama.ConsumerMessage, t model.TrackModel) error {
	// 更新mysql
	err := r.trackDao.UpdateCollect(t.TrackId, t.IsDeleted)
	if err != nil {
		zap.L().Error("Consume Insert error:", zap.Error(err))
		return err
	}
	// 更新es
	//err = client.PushCollect(r.client, &t)
	//if err != nil {
	//	zap.L().Error("Consume Insert error:", zap.Error(err))
	//	return err
	//}
	return nil
}
func NewChatConsumers(consumer *TrackConsumer) []Consumer {
	KafkaConsumers = append(KafkaConsumers, consumer)
	return KafkaConsumers
}
